package org.pocs.akka.untyped;

import java.util.ArrayList;
import java.util.List;

import org.pocs.common.dto.Signal;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.routing.BroadcastRouter;

public class AkkaInitiator extends UntypedActor {

	static ActorSystem system = null;
	
	static {
		
		system = ActorSystem.create("JavaSimulator");
	}
	
	private int responseCount = 0;
	private int signalCount = 0;
	
	public void processSignal(List<Signal> signalList) {
		
		List<ActorRef> actorRefs = new ArrayList<>();
		ActorRef actorRef = null;
		
		for (Signal signal : signalList) {
			actorRef = getContext().actorOf(WorkerActor.getProps(signal));
			actorRefs.add(actorRef);
		}
		
		signalCount = signalList.size();
		
		ActorRef router = getContext().actorOf(Props.empty().withRouter(BroadcastRouter.create(actorRefs)));
		router.tell(CurrentStatus.PROCESS_SIGNAL, getSelf());
	}
	
	public static void main(String[] args) {
		
	}

	@Override
	public void onReceive(Object msg) throws Exception {
		
		if(msg instanceof CurrentStatus) {
			CurrentStatus status = (CurrentStatus) msg;
			
			switch (status) {
			case FETCH_DATA:
				// Connect to Redis to get row-by-row data
				List<Signal> signalList = new ArrayList<>();
				
				// Fill from Redis
				
				processSignal(signalList);
				break;
			case ACTOR_FAIL:
				responseCount += 1;
				if(responseCount >= signalCount) {
					getSelf().tell(CurrentStatus.FETCH_DATA, getSelf());
				}
				
				break;
			case ACTOR_SUCCESS:
				responseCount += 1;
                if(responseCount >= signalCount) {
                	getSelf().tell(CurrentStatus.FETCH_DATA, getSelf());
				}
				
				break;
			}	
		}
	}
	
	public static void initializeSimulator() {
		// Activate Akka Initiator
			ActorRef masterRef = system.actorOf(Props.create(AkkaInitiator.class));
			masterRef.tell(CurrentStatus.FETCH_DATA, masterRef);	
		
	}
}